/**
 * Copyright (C) 2010 Neofonie GmbH
 *
 * This programm is free software; you can redistribute it and/or modify
 * it under the terms of the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package eu.dicodeproject.analysis.hashtags;

import java.io.IOException;
import java.text.FieldPosition;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.Locale;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.mahout.common.StringTuple;

import eu.dicodeproject.analysis.histogram.HBaseHistogramMapper.ErrorCases;

/**
 * Extracts hashtags and writes day/hashtag-pairs and counts to HDFS.
 */
public class HashtagsTimelineMapper extends TableMapper<Text, StringTuple> {
	
	  /** Set of patterns with associated key for later output. */
	  public static enum ErrorCases {
	    TWEETNULL, DATENULL
	  };

	  /** The year we started tracking twitter - our year zero. */
	  private static final int TRACKING_START_YEAR = 2010;
	  /** "Average" amount of days in a year. Ignoring leap years etc for now. */
	  private static final int DAYS_IN_YEAR = 365;

	  /** Date formatters for x creation. */
	  private SimpleDateFormat dateFormat;
	  private SimpleDateFormat dayOfYear;
	  private SimpleDateFormat year;
	  
	  private Text dateSpaceHashtag = new Text();
	  private final static IntWritable ONE = new IntWritable(1);
	  private Pattern pattern = Pattern.compile("\\s+#\\w+");


	  @Override
	  protected void setup(Context context) throws IOException, InterruptedException {
	    super.setup(context);
	    // Wed, 15 Dec 2010 09:39:23 +0000
	    this.dateFormat = new SimpleDateFormat("EEE, dd MMM yyyy HH:mm:ss Z", Locale.ENGLISH);
	    this.dayOfYear = new SimpleDateFormat("DDD", Locale.ENGLISH);
	    this.year = new SimpleDateFormat("yyyy", Locale.ENGLISH);

	  }

	

	@Override
	protected void map(ImmutableBytesWritable row, Result values,
			Mapper.Context context) throws IOException, InterruptedException {
		
		// extract the data
	    byte[] creationDateBytes = values.getValue(Bytes.toBytes("metaFamily"), Bytes.toBytes("creationDate"));
	    byte[] tweetTextBytes = values.getValue(Bytes.toBytes("textFamily"), Bytes.toBytes("text"));
	    byte[] hashtagBytes = null;
	    Calendar cal = Calendar.getInstance();


	    // we anticipate errors as in non-existant tweetText or creationDate - those
	    // are ignored
	    boolean failed = false;
	    if (tweetTextBytes == null) {
	      context.getCounter(ErrorCases.TWEETNULL).increment(1);
	      failed = true;
	    }
	    if (creationDateBytes == null) {
	      context.getCounter(ErrorCases.DATENULL).increment(1);
	      failed = true;
	    }

	    if (!failed) {

	        String creationDate = new String(creationDateBytes);
	        // TODO: extract to validation method 
	        if (creationDate.endsWith("+000")){
	        	creationDate+= "0";
	        }
	        Date date = null;
	        try {
	  	date = this.dateFormat.parse(creationDate);
	        } catch (ParseException e) {
	  	throw new IOException(e);
	        }

	        StringBuffer dayOfYearBuf = new StringBuffer();
	        this.dayOfYear.format(date, dayOfYearBuf, new FieldPosition(0));
	        int day = Integer.parseInt(dayOfYearBuf.toString());

	        StringBuffer yearBuf = new StringBuffer();
	        this.year.format(date, yearBuf, new FieldPosition(0));
	        int year = Integer.parseInt(yearBuf.toString()) - TRACKING_START_YEAR;

	        int x = year * DAYS_IN_YEAR + day;

	        // extract hashtags
	    	String value = new String(tweetTextBytes);
			Matcher matcher = pattern.matcher(value);

			while (matcher.find()) {
				dateSpaceHashtag.set(x+" "); // TODO: remove string concatenation!
				hashtagBytes = Bytes.toBytes(matcher.group().toLowerCase().trim());
				dateSpaceHashtag.append(hashtagBytes, 0, hashtagBytes.length);
				context.write(dateSpaceHashtag, ONE);
			}
		}
	}
}
